/*!
 * \file rocketmq_core.c
 *
 * \brief 共通定义
 *
 * All Rights Reserved Copyright (C) 2014 broada,Co.,Ltd.\n
 * 共通使用的宏和结构体
 *
 * \version 1.0
 * \author jikw
 * \date 2017/4/10
 *
 */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <json-c/json.h>

#include "config.h"
#include "frct_comm.h"
#include "frreport_comm.h"
#include "frservice_comm.h"
#include "frbyte_buffer.h"
#include "iconv2utf8.h"
#include "rocketmq_core.h"

extern void *g_p_apm_context;
struct timeval g_st_time;

typedef struct __st_rocketmq_data
{
    L4_transaction st_l4_trans;                 //L4信息统计
    Frlist_header *p_report_comm;               //用于临时存储report的公共部分数据
    Frlist_header *p_report_req;                //用于临时存储report的请求部分数据
} rocketmq_data;

typedef struct __st_L7session_rocketmq_private
{
    unsigned long ul_seq;                       //序列号，1次req和response时，使用同一个序列号
    L4_transaction st_l4_trans;                 //数据
    Frreport_body_data *p_st_body_data_req;
    Frreport_body_data *p_st_body_data_resp;
    Frbyte_Packet *p_req_packet;                //存请求bytebuff数据
    Frbyte_Packet *p_resp_packet;               //存相应bytebuff数据
    Frhash_header *p_rocketmq_list;             //存rocketmq_data
} L7session_rocketmq_private;

typedef struct __st_rocketmq_map
{
    unsigned int ui_code;
    char arr_name[128];
} rocketmq_map;

rocketmq_map rocketmq_reqcode_map[] = {
    {10, "Broker 发送消息"},
    {11, "Broker 订阅消息"},
    {12, "Broker 查询消息"},
    {13, "Broker 查询Broker Offset"},
    {14, "Broker 查询Consumer Offset"},
    {15, "Broker 更新Consumer Offset"},
    {17, "Broker 更新或者增加一个Topic"},
    {21, "Broker 获取所有Topic的配置（Slave和Namesrv都会向Master请求此配置）"},
    {22, "Broker 获取所有Topic配置（Slave和Namesrv都会向Master请求此配置）"},
    {23, "Broker 获取所有Topic名称列表"},
    {25, "Broker 更新Broker上的配置"},
    {26, "Broker 获取Broker上的配置"},
    {27, "Broker 触发Broker删除文件"},
    {28, "Broker 获取Broker运行时信息"},
    {29, "Broker 根据时间查询队列的Offset"},
    {30, "Broker 查询队列最大Offset"},
    {31, "Broker 查询队列最小Offset"},
    {32, "Broker 查询队列最早消息对应时间"},
    {33, "Broker 根据消息ID来查询消息"},
    {34, "Broker Client向Client发送心跳，并注册自身"},
    {35, "Broker Client注销"},
    {36, "Broker Consumer将处理不了的消息发回服务器"},
    {37, "Broker Commit或者Rollback事务"},
    {38, "Broker 获取ConsumerId列表通过GroupName"},
    {39, "Broker 主动向Producer回查事务状态"},
    {40, "Broker Broker通知Consumer列表变化"},
    {41, "Broker Consumer向Master锁定队列"},
    {42, "Broker Consumer向Master解锁队列"},
    {43, "Broker 获取所有Consumer Offset"},
    {45, "Broker 获取所有定时进度"},
    {100, "Namesrv 向Namesrv追加KV配置"},
    {101, "Namesrv 从Namesrv获取KV配置"},
    {102, "Namesrv 从Namesrv获取KV配置"},
    {103, "Namesrv 注册一个Broker，数据都是持久化的，如果存在则覆盖配置"},
    {104, "Namesrv 卸载一个Broker，数据都是持久化的"},
    {105, "Namesrv 根据Topic获取Broker Name、队列数(包含读队列与写队列)"},
    {106, "Namesrv 获取注册到Name Server的所有Broker集群信息"},
    {200, "Namesrv 获取注册到Name Server的所有Broker集群信息"},
    {201, "Namesrv 获取注册到Name Server的所有Broker集群信息"},
    {202, "Namesrv 获取注册到Name Server的所有Broker集群信息"},
    {203, "Namesrv 获取注册到Name Server的所有Broker集群信息"},
    {204, "Namesrv 获取注册到Name Server的所有Broker集群信息"},
    {205, "Namesrv 获取注册到Name Server的所有Broker集群信息"},
    {206, "从Name Server获取完整Topic列表"},
    {207, "从Broker删除订阅组"},
    {208, "从Broker获取消费状态（进度）"},
    {209, "Suspend Consumer消费过程"},
    {210, "Resume Consumer消费过程"},
    {211, "重置Consumer Offset"},
    {212, "重置Consumer Offset"},
    {213, "调整Consumer线程池数量"},
    {214, "查询消息被哪些消费组消费"},
    {215, "从Broker删除Topic配置"},
    {216, "从Namesrv删除Topic配置"},
    {217, "Namesrv 通过 project 获取所有的 server ip 信息"},
    {218, "Namesrv 删除指定 project group 下的所有 server ip 信息"},
    {219, "通过NameSpace获取所有的KV List"},
    {220, "offset 重置"},
    {221, "客户端订阅消息"},
    {222, "通知 broker 调用 offset 重置处理"},
    {223, "通知 broker 调用客户端订阅消息处理"},
    {224, "获取指定集群下的所有 topic"},
    {300, "Broker 查询topic被谁消费"},
    {301, "向Broker注册Filter Server"},
    {302, "向Filter Server注册Class"},
    {303, "根据 topic 和 group 获取消息的时间跨度"}
};

rocketmq_map rocketmq_respcode_map[] = {
    {10, "Broker 刷盘超时"},
    {11, "Broker 同步双写，Slave不可用"},
    {12, "Broker 同步双写，等待Slave应答超时"},
    {13, "Broker 消息非法"},
    {14, "Broker, Namesrv 服务不可用，可能是正在关闭或者权限问题"},
    {15, "Broker, Namesrv 版本号不支持"},
    {16, "Broker, Namesrv 无权限执行此操作，可能是发、收、或者其他操作"},
    {17, "Broker, Topic不存在"},
    {18, "Broker, Topic已经存在，创建Topic"},
    {19, "Broker 拉消息未找到（请求的Offset等于最大Offset，最大Offset无对应消息）"},
    {20, "Broker 可能被过滤，或者误通知等"},
    {21, "Broker 拉消息请求的Offset不合法，太小或太大"},
    {22, "Broker 查询消息未找到"},
    {23, "Broker 订阅关系解析失败"},
    {24, "Broker 订阅关系不存在"},
    {25, "Broker 订阅关系不是最新的"},
    {26, "Broker 订阅组不存在"},
    {200, "Producer 事务应该被提交"},
    {201, "Producer 事务应该被回滚"},
    {202, "Producer 事务状态未知"},
    {203, "Producer ProducerGroup错误"},
    {204, "单元化消息，需要设置 buyerId"},
    {205, "单元化消息，非本单元消息"},
    {206, "Consumer不在线"}
};

rocketmq_map rocketmq_version_map[] = {
    {1, "V3_0_0_SNAPSHOT"},
    {2, "V3_0_0_ALPHA1"},
    {3, "V3_0_0_BETA1"},
    {4, "V3_0_0_BETA2"},
    {5, "V3_0_0_BETA3"},
    {6, "V3_0_0_BETA4"},
    {7, "V3_0_0_BETA5"},
    {8, "V3_0_0_BETA6_SNAPSHOT"},
    {9, "V3_0_0_BETA6"},
    {10, "V3_0_0_BETA7_SNAPSHOT"},
    {11, "V3_0_0_BETA7"},
    {12, "V3_0_0_BETA8_SNAPSHOT"},
    {13, "V3_0_0_BETA8"},
    {14, "V3_0_0_BETA9_SNAPSHOT"},
    {15, "V3_0_0_BETA9"},
    {16, "V3_0_0_FINAL"},
    {17, "V3_0_1_SNAPSHOT"},
    {18, "V3_0_1"},
    {19, "V3_0_2_SNAPSHOT"},
    {20, "V3_0_2"},
    {21, "V3_0_3_SNAPSHOT"},
    {22, "V3_0_3"},
    {23, "V3_0_4_SNAPSHOT"},
    {24, "V3_0_4"},
    {25, "V3_0_5_SNAPSHOT"},
    {26, "V3_0_5"},
    {27, "V3_0_6_SNAPSHOT"},
    {28, "V3_0_6"},
    {29, "V3_0_7_SNAPSHOT"},
    {30, "V3_0_7"},
    {31, "V3_0_8_SNAPSHOT"},
    {32, "V3_0_8"},
    {33, "V3_0_9_SNAPSHOT"},
    {34, "V3_0_9"},
    {35, "V3_0_10_SNAPSHOT"},
    {36, "V3_0_10"},
    {37, "V3_0_11_SNAPSHOT"},
    {38, "V3_0_11"},
    {39, "V3_0_12_SNAPSHOT"},
    {40, "V3_0_12"},
    {41, "V3_0_13_SNAPSHOT"},
    {42, "V3_0_13"},
    {43, "V3_0_14_SNAPSHOT"},
    {44, "V3_0_14"},
    {45, "V3_0_15_SNAPSHOT"},
    {46, "V3_0_15"},
    {47, "V3_1_0_SNAPSHOT"},
    {48, "V3_1_0"},
    {49, "V3_1_1_SNAPSHOT"},
    {50, "V3_1_1"},
    {51, "V3_1_2_SNAPSHOT"},
    {52, "V3_1_2"},
    {53, "V3_1_3_SNAPSHOT"},
    {54, "V3_1_3"},
    {55, "V3_1_4_SNAPSHOT"},
    {56, "V3_1_4"},
    {57, "V3_1_5_SNAPSHOT"},
    {58, "V3_1_5"},
    {59, "V3_1_6_SNAPSHOT"},
    {60, "V3_1_6"},
    {61, "V3_1_7_SNAPSHOT"},
    {62, "V3_1_7"},
    {63, "V3_1_8_SNAPSHOT"},
    {64, "V3_1_8"},
    {65, "V3_1_9_SNAPSHOT"},
    {66, "V3_1_9"},
    {67, "V3_2_0_SNAPSHOT"},
    {68, "V3_2_0"},
    {69, "V3_2_1_SNAPSHOT"},
    {70, "V3_2_1"},
    {71, "V3_2_2_SNAPSHOT"},
    {72, "V3_2_2"},
    {73, "V3_2_3_SNAPSHOT"},
    {74, "V3_2_3"},
    {75, "V3_2_4_SNAPSHOT"},
    {76, "V3_2_4"},
    {77, "V3_2_5_SNAPSHOT"},
    {78, "V3_2_5"},
    {79, "V3_2_6_SNAPSHOT"},
    {80, "V3_2_6"},
    {81, "V3_2_7_SNAPSHOT"},
    {82, "V3_2_7"},
    {83, "V3_2_8_SNAPSHOT"},
    {84, "V3_2_8"},
    {85, "V3_2_9_SNAPSHOT"},
    {86, "V3_2_9"},
    {87, "V3_3_1_SNAPSHOT"},
    {88, "V3_3_1"},
    {89, "V3_3_2_SNAPSHOT"},
    {90, "V3_3_2"},
    {91, "V3_3_3_SNAPSHOT"},
    {92, "V3_3_3"},
    {93, "V3_3_4_SNAPSHOT"},
    {94, "V3_3_4"},
    {95, "V3_3_5_SNAPSHOT"},
    {96, "V3_3_5"},
    {97, "V3_3_6_SNAPSHOT"},
    {98, "V3_3_6"},
    {99, "V3_3_7_SNAPSHOT"},
    {100, "V3_3_7"},
    {101, "V3_3_8_SNAPSHOT"},
    {102, "V3_3_8"},
    {103, "V3_3_9_SNAPSHOT"},
    {104, "V3_3_9"}
};

/*!
 * rocketmq私有数据创建
 *
 * \param void
 *
 * \return NULL          内存分配失败\n
 *         p_st_private  内存分配成功
 *
 * \note
 *
 */
FR_STATIC L7session_rocketmq_private* __rocketmq_private_create(void)
{
    L7session_rocketmq_private *p_st_private = NULL;
    p_st_private = (L7session_rocketmq_private*)frmalloc(sizeof(L7session_rocketmq_private));
    if (p_st_private == NULL)
        return NULL;
    memset(p_st_private, 0x00, sizeof(L7session_rocketmq_private));

    return p_st_private;
}

/*!
 * 私有信息重置
 *
 * \param p_private_session  (rocketmq的私有session数据)
 *
 * \return FR_SUCC   成功\n
 *
 * \note
 *
 */
FR_STATIC int __rocketmq_session_reset(void *p_private_session)
{
    L7session_rocketmq_private *p_st_private = (L7session_rocketmq_private*)p_private_session;

    if (p_st_private)
    {
        if (p_st_private->p_st_body_data_req)
            frreport_body_free(&p_st_private->p_st_body_data_req);

        if (p_st_private->p_st_body_data_resp)
            frreport_body_free(&p_st_private->p_st_body_data_resp);

        if (p_st_private->p_req_packet)
            bytebuff_free(&(p_st_private->p_req_packet));

        if (p_st_private->p_resp_packet)
            bytebuff_free(&(p_st_private->p_resp_packet));

        memset(&p_st_private->st_l4_trans, 0x00, sizeof(L4_transaction));
    }
    return FR_SUCC;
}

/*!
 * 用二分法查找rocketmq_reqcode_map匹配的序号
 *
 * \param i_key           (code)
 *
 * \return -1              未找到\n
 *         index           找到的下标
 *
 * \note
 *
 */
FR_STATIC int __get_reqcode(unsigned int i_key)
{
    int i_len = 0;
    int i_mid = 0;
    int i_front = 0;
    int i_back = 0;

    i_len = sizeof(rocketmq_reqcode_map) / sizeof(rocketmq_reqcode_map[0]);
    i_back = i_len - 1;

    while (i_front <= i_back)
    {
        i_mid = (i_front + i_back) / 2;
        if (rocketmq_reqcode_map[i_mid].ui_code == i_key)
            return i_mid;
        if (rocketmq_reqcode_map[i_mid].ui_code < i_key)
            i_front = i_mid + 1;
        else
            i_back = i_mid - 1;
    }

    return -1;
}

/*!
 * 用二分法查找rocketmq_respcode_map匹配的序号
 *
 * \param i_key           (code)
 *
 * \return -1              未找到\n
 *         index           找到的下标
 *
 * \note
 *
 */
FR_STATIC int __get_respcode(unsigned int i_key)
{
    int i_len = 0;
    int i_mid = 0;
    int i_front = 0;
    int i_back = 0;

    i_len = sizeof(rocketmq_respcode_map) / sizeof(rocketmq_respcode_map[0]);
    i_back = i_len - 1;

    while (i_front <= i_back)
    {
        i_mid = (i_front + i_back) / 2;
        if (rocketmq_respcode_map[i_mid].ui_code == i_key)
            return i_mid;
        if (rocketmq_respcode_map[i_mid].ui_code < i_key)
            i_front = i_mid + 1;
        else
            i_back = i_mid - 1;
    }

    return -1;
}

/*!
 * 用二分法查找rocketmq_version_map匹配的序号
 *
 * \param i_key           (code)
 *
 * \return -1              未找到\n
 *         index           找到的下标
 *
 * \note
 *
 */
FR_STATIC int __get_version(unsigned int i_key)
{
    int i_len = 0;
    int i_mid = 0;
    int i_front = 0;
    int i_back = 0;

    i_len = sizeof(rocketmq_version_map) / sizeof(rocketmq_version_map[0]);
    i_back = i_len - 1;

    while (i_front <= i_back)
    {
        i_mid = (i_front + i_back) / 2;
        if (rocketmq_version_map[i_mid].ui_code == i_key)
            return i_mid;
        if (rocketmq_version_map[i_mid].ui_code < i_key)
            i_front = i_mid + 1;
        else
            i_back = i_mid - 1;
    }

    return -1;
}

/*!
 * 判断是否是rocketmq协议
 *
 * \param p_st_conn      (输入数据信息)
 * \param p_st_packet    (输入数据)
 *
 * \return APM_L7_IS_YES  属于rocketmq协议\n
 *         APM_L7_IS_NO   不属于rocketmq协议
 *
 * \note
 *
 */
int rocketmq_detect(Conntrack *p_st_conn, Apm_packet *p_st_packet)
{
    unsigned int i_ret = 0;
    unsigned int i_len = 0;
    int i_code = 0;
    int i_flag = 0;
    int i_header_len = 0;
    char arr_len[8];
    char arr_header_len[8];

    char *p_value = NULL;
    json_object *p_header = NULL;
    json_object *p_code = NULL;
    json_object *p_flag = NULL;

    if (p_st_conn==NULL ||
        p_st_packet == NULL ||
        p_st_packet->p_payload == NULL ||
        p_st_packet->ui_payload_len < 8)
        return APM_L7_IS_NO;

    //|<-- 4 -->|<----- 4 ----->|<----------->|<--------->|
    //| length  | header length | header data | body data |

    //length：大端4个字节整数，等于2、3、4长度总和，但是有分片包情况
    memcpy(arr_len, p_st_packet->p_payload, 4);
    arr_len[4] = '\0';
    i_len = ntohl(*(int *)arr_len);

    //header length：大端4个字节整数，等于3的长度
    memcpy(arr_header_len, p_st_packet->p_payload+4, 4);
    arr_header_len[4] = '\0';
    i_header_len = ntohl(*(int *)arr_header_len);

    //Header data：使用json序列化数据
    p_value = (char *)frmalloc(i_header_len+1);
    if (p_value == NULL)
        goto ERROR;
    memcpy(p_value, p_st_packet->p_payload+8, i_header_len);
    p_value[i_header_len] = '\0';
    p_header = json_tokener_parse(p_value);
    if (p_header == NULL)
        goto ERROR;
    //flag：通信局的标志位
    p_flag = json_object_object_get(p_header, "flag");
    if (p_flag == NULL)
        goto ERROR;
    i_flag = json_object_get_int(p_flag);
    //code：请求操作代码/应答结果代码
    p_code = json_object_object_get(p_header, "code");
    if (p_code == NULL)
        goto ERROR;
    i_code = json_object_get_int(p_code);

    if (i_flag == 0)
    {
        if (__get_reqcode(i_code) == -1)
            goto ERROR;

        i_ret = frct_get_direct_known(p_st_conn);
        if (i_ret == FR_DIRECT_UNKNOWN)
            frcap_set_cs_srv(p_st_conn, p_st_packet->dst_ip, p_st_packet->dst_port);
        frservice_add(g_p_apm_context, MODULE_L7PROTO, p_st_packet->dst_ip, p_st_packet->dst_port, NULL, 0);
    }
    else if (i_flag == 1)
    {
        if (i_code != 0 || __get_respcode(i_code) == -1)
            goto ERROR;

        i_ret = frct_get_direct_known(p_st_conn);
        if (i_ret == FR_DIRECT_UNKNOWN)
            frcap_set_cs_srv(p_st_conn, p_st_packet->src_ip, p_st_packet->src_port);
        frservice_add(g_p_apm_context, MODULE_L7PROTO, p_st_packet->src_ip, p_st_packet->src_port, NULL, 0);
    }
    else
        goto ERROR;

    frfree(p_value);
    json_object_put(p_header);
    return APM_L7_IS_YES;

ERROR:
    if (p_value)
        frfree(p_value);
    if (p_header)
        json_object_put(p_header);
    //context 查询
    i_ret = frservice_get(g_p_apm_context, MODULE_L7PROTO, p_st_packet->dst_ip, p_st_packet->dst_port, NULL, 0, NULL);
    if (i_ret == FR_SUCC)
    {
        if (frct_get_direct_known(p_st_conn) == FR_DIRECT_UNKNOWN)
            frcap_set_cs_srv(p_st_conn, p_st_packet->dst_ip, p_st_packet->dst_port);

        return APM_L7_IS_YES;
    }

    i_ret = frservice_get(g_p_apm_context, MODULE_L7PROTO, p_st_packet->src_ip, p_st_packet->src_port, NULL, 0, NULL);
    if (i_ret == FR_SUCC)
    {
        if (frct_get_direct_known(p_st_conn) == FR_DIRECT_UNKNOWN)
            frcap_set_cs_srv(p_st_conn, p_st_packet->src_ip, p_st_packet->src_port);

        return APM_L7_IS_YES;
    }

    return APM_L7_IS_NO;
}

/*!
 * 链表节点释放规则
 *
 * \Frlist_node *p_st_node
 *
 * \return void
 *
 * \note
 *
 */
void rocketmq_list_node_free(Frlist_node *p_st_node)
{
    rocketmq_data *p_st_info = NULL;

    if (p_st_node==NULL || p_st_node->data == NULL )
        return;

    p_st_info = (rocketmq_data *)FRLIST_GET_NODE_DATA(p_st_node);

    if (p_st_info->p_report_req)
        frreport_body_l7list_free(&p_st_info->p_report_req);
    if (p_st_info->p_report_comm)
        frreport_body_l7list_free(&p_st_info->p_report_comm);

    frfree(p_st_info);
    return;
}

/*!
 * req数据解析
 *
 * \param p_packet        (req数据)
 * \param p_st_body_data
 * \param p_i_opaque      (req异步信息)
 *
 * \return APM_L7_ANALYZE_SUCC   成功\n
 *         APM_L7_ANALYZE_FAIL   失败
 *
 * \note
 *
 */
FR_STATIC int __req_parse(Frbyte_Packet *p_packet, Frreport_body_data *p_st_body_data, int *p_i_opaque)
{
    uint32_t u32_len = 0;
    uint32_t u32_header_len = 0;
    int i_data = 0;
    int i_code = 0;
    int i_index = 0;
    int i_version = 0;
    char *p_header = NULL;
    char *p_data = NULL;
    json_object *p_json = NULL;
    json_object *p_opaque = NULL;
    json_object *p_code = NULL;
    json_object *p_version = NULL;
    json_object *p_language = NULL;
    json_object *p_remark = NULL;
    json_object *p_extFields = NULL;

    if (p_packet == NULL || p_st_body_data == NULL || p_i_opaque == NULL)
        return APM_L7_ANALYZE_FAIL;

    //|<-- 4 -->|<----- 4 ----->|<----------->|<--------->|
    //| length  | header length | header data | body data |

    //length：大端4个字节整数，等于2、3、4长度总和
    if (bytebuff_get_int(p_packet, 4, &u32_len, BYTE_BIG_ENDIAN) == FR_FAIL)
        goto ERROR;

    //header length：大端4个字节整数，等于3的长度
    if (bytebuff_get_int(p_packet, 4, &u32_header_len, BYTE_BIG_ENDIAN) == FR_FAIL)
        goto ERROR;

    //Header data：使用json序列化数据
    p_header = (char *)frmalloc(u32_header_len+1);
    if (p_header == NULL)
        goto ERROR;
    if (bytebuff_get_bytes(p_packet, p_header, u32_header_len) == FR_FAIL)
        goto ERROR;
    p_header[u32_header_len] = '\0';
    frreport_body_l7_add_str(p_st_body_data, "request.rocketmq.header", p_header);
    //code：整数，请求操作代码，请求接收方根据不同的代码做不同的操作
    //language：字符串，请求发起方实现语言，默认JAVA
    //version：整数，请求发起方程序版本
    //opaque：整数，请求发起方在同一连接上不同的请求标识代码，多线程连接复用使用
    //flag：整数，通信层的标志位
    //remark：字符串，传输自定义文本信息
    //extFields：HashMap<String,String>，请求自定义字段
    //opaque
    p_json = json_tokener_parse(p_header);
    if (p_json == NULL)
        goto ERROR;
    p_opaque = json_object_object_get(p_json, "opaque");
    if (p_opaque == NULL)
        goto ERROR;
    *p_i_opaque = json_object_get_int(p_opaque);
    //code
    p_code = json_object_object_get(p_json, "code");
    if (p_code == NULL)
        goto ERROR;
    i_code = json_object_get_int(p_code);
    frreport_body_l7_add_int(p_st_body_data, "transaction.type_code", i_code);
    if ((i_index = __get_reqcode(i_code)) == -1)
        goto ERROR;
    frreport_body_l7_add_str(p_st_body_data, "transaction.type_name", rocketmq_reqcode_map[i_index].arr_name);
    frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_TRANS_TITLE, rocketmq_reqcode_map[i_index].arr_name);
    //version
    p_version = json_object_object_get(p_json, "version");
    if (p_version == NULL)
        goto ERROR;
    i_version = json_object_get_int(p_version);
    if ((i_index = __get_version(i_version)) == -1)
        goto ERROR;
    frreport_body_l7_add_str(p_st_body_data, "request.rocketmq.version", rocketmq_version_map[i_index].arr_name);
    //language
    p_language = json_object_object_get(p_json, "language");
    if (p_language == NULL)
        goto ERROR;
    frreport_body_l7_add_str(p_st_body_data, "request.rocketmq.language", json_object_get_string(p_language));
    //remark
    p_remark = json_object_object_get(p_json, "remark");
    if (p_remark != NULL)
        frreport_body_l7_add_str(p_st_body_data, "request.rocketmq.remark", json_object_get_string(p_remark));
    //extFields
    p_extFields = json_object_object_get(p_json, "extFields");
    if (p_extFields == NULL)
        goto ERROR;
    frreport_body_l7_add_str(p_st_body_data, "request.rocketmq.extFields", json_object_get_string(p_extFields));

    //body data：应用自定义二进制序列化数据
    if (p_packet->ui_remaining == 0)
    {
        json_object_put(p_json);
        frfree(p_header);
        return APM_L7_ANALYZE_SET_REQ(APM_L7_ANALYZE_SUCC);
    }
    i_data = u32_len-u32_header_len-4;
    p_data = (char *)frmalloc(i_data+1);
    if (p_data == NULL)
        goto ERROR;
    if (bytebuff_get_bytes(p_packet, p_data, i_data) == FR_FAIL)
        goto ERROR;
    p_data[i_data] = '\0';
    frreport_body_l7_add_str(p_st_body_data, "request.rocketmq.body", p_data);

    json_object_put(p_json);
    frfree(p_header);
    frfree(p_data);
    return APM_L7_ANALYZE_SET_REQ(APM_L7_ANALYZE_SUCC);

ERROR:
    if (p_json)
        json_object_put(p_json);
    if (p_header)
        frfree(p_header);
    if (p_data)
        frfree(p_data);
    return APM_L7_ANALYZE_FAIL;
}

/*!
 * resp数据解析
 *
 * \param p_packet        (resp数据)
 * \param p_st_body_data
 * \param p_i_opaque      (resp异步信息)
 *
 * \return APM_L7_ANALYZE_SET_RESP(APM_L7_ANALYZE_SUCC)   成功\n
 *         APM_L7_ANALYZE_FAIL   失败
 *
 * \note
 *
 */
FR_STATIC int __resp_parse(Frbyte_Packet *p_packet, Frreport_body_data *p_st_body_data, int *p_i_opaque)
{
    uint32_t u32_len = 0;
    uint32_t u32_header_len = 0;
    int i_data = 0;
    int i_code = 0;
    int i_index = 0;
    int i_version = 0;
    char *p_header = NULL;
    char *p_data = NULL;
    json_object *p_json = NULL;
    json_object *p_opaque = NULL;
    json_object *p_code = NULL;
    json_object *p_version = NULL;
    json_object *p_language = NULL;
    json_object *p_remark = NULL;
    json_object *p_extFields = NULL;

    if (p_packet == NULL || p_st_body_data == NULL || p_i_opaque == NULL)
        return APM_L7_ANALYZE_FAIL;

    //|<-- 4 -->|<----- 4 ----->|<----------->|<--------->|
    //| length  | header length | header data | body data |

    //length：大端4个字节整数，等于2、3、4长度总和
    if (bytebuff_get_int(p_packet, 4, &u32_len, BYTE_BIG_ENDIAN) == FR_FAIL)
        goto ERROR;

    //header length：大端4个字节整数，等于3的长度
    if (bytebuff_get_int(p_packet, 4, &u32_header_len, BYTE_BIG_ENDIAN) == FR_FAIL)
        goto ERROR;

    //Header data：使用json序列化数据
    p_header = (char *)frmalloc(u32_header_len+1);
    if (p_header == NULL)
        goto ERROR;
    if (bytebuff_get_bytes(p_packet, p_header, u32_header_len) == FR_FAIL)
        goto ERROR;
    p_header[u32_header_len] = '\0';
    frreport_body_l7_add_str(p_st_body_data, "response.rocketmq.header", p_header);
    //code：整数，应答结果代码，0表示成功，非0表示各种错误代码
    //language：字符串，应答接收方实现语言
    //version：整数，应答接收方程序版本
    //opaque：整数，应答方不做修改，直接返回
    //flag：整数，通信层的标志位
    //remark：字符串，错误详细描述信息
    //extFields：HashMap<String,String>，应答自定义字段
    p_json = json_tokener_parse(p_header);
    if (p_json == NULL)
        goto ERROR;
    //opaque
    p_opaque = json_object_object_get(p_json, "opaque");
    if (p_opaque == NULL)
        goto ERROR;
    *p_i_opaque = json_object_get_int(p_opaque);
    //code
    p_code = json_object_object_get(p_json, "code");
    if (p_code == NULL)
        goto ERROR;
    i_code = json_object_get_int(p_code);
    frreport_body_l7_add_int(p_st_body_data, "response.code", i_code);
    if (i_code == 0)
        frreport_body_l7_add_str(p_st_body_data, "response.code_string", "成功");
    else
    {
        if ((i_index = __get_respcode(i_code)) == -1)
            goto ERROR;
        frreport_body_l7_add_str(p_st_body_data, "response.code_string", rocketmq_respcode_map[i_index].arr_name);
    }
    //version
    p_version = json_object_object_get(p_json, "version");
    if (p_version == NULL)
        goto ERROR;
    i_version = json_object_get_int(p_version);
    if ((i_index = __get_version(i_version)) == -1)
        goto ERROR;
    frreport_body_l7_add_str(p_st_body_data, "response.rocketmq.version", rocketmq_version_map[i_index].arr_name);
    //language
    p_language = json_object_object_get(p_json, "language");
    if (p_language == NULL)
        goto ERROR;
    frreport_body_l7_add_str(p_st_body_data, "response.rocketmq.language", json_object_get_string(p_language));
    //remark
    p_remark = json_object_object_get(p_json, "remark");
    if (p_remark != NULL)
        frreport_body_l7_add_str(p_st_body_data, "response.error_message", json_object_get_string(p_remark));
    //extFields
    p_extFields = json_object_object_get(p_json, "extFields");
    if (p_extFields == NULL)
        goto ERROR;
    frreport_body_l7_add_str(p_st_body_data, "response.rocketmq.extFields", json_object_get_string(p_extFields));

    //body data：应用自定义二进制序列化数据
    if (p_packet->ui_remaining == 0)
    {
        json_object_put(p_json);
        frfree(p_header);
        return APM_L7_ANALYZE_SET_RESP(APM_L7_ANALYZE_SUCC);
    }
    i_data = u32_len-u32_header_len-4;
    p_data = (char *)frmalloc(i_data+1);
    if (p_data == NULL)
        goto ERROR;
    if (bytebuff_get_bytes(p_packet, p_data, i_data) == FR_FAIL)
        goto ERROR;
    p_data[i_data] = '\0';
    frreport_body_l7_add_str(p_st_body_data, "response.rocketmq.body", p_data);

    json_object_put(p_json);
    frfree(p_header);
    frfree(p_data);
    return APM_L7_ANALYZE_SET_RESP(APM_L7_ANALYZE_SUCC);

ERROR:
    if (p_json)
        json_object_put(p_json);
    if (p_header)
        frfree(p_header);
    if (p_data)
        frfree(p_data);
    return APM_L7_ANALYZE_FAIL;
}

/*!
 * req的p_report_comm、p_report_req、L4信息和ID存进链表
 *
 * \param p_st_private    (rocketmq私有数据)
 * \param p_mesgid        (mesgID)
 *
 * \return FR_FAIL        失败\n
 *         FR_SUCC        成功
 *
 * \note
 *
 */
FR_STATIC int __rocketmq_add_req_list(L7session_rocketmq_private *p_st_private, int i_opaque)
{
    rocketmq_data *p_st_report = NULL;
	Frlist_node *p_st_node = NULL;
	Frlist_node *p_st_gotnode = NULL;
	L4_transaction *p_st_l4_trans = NULL;

    if (p_st_private == NULL)
        return FR_FAIL;

    //创建链表
    if (p_st_private->p_rocketmq_list == NULL)
        p_st_private->p_rocketmq_list = frhash2_create2(ROCKETMQ_HASH_SIZE);

    p_st_l4_trans = &(p_st_private->st_l4_trans);

    //req的p_report_comm、p_report_req、L4信息和ID存进链表
    p_st_report = (rocketmq_data *)frmalloc(sizeof(rocketmq_data));
    memset(p_st_report, 0x00, sizeof(rocketmq_data));

    //拷贝L4信息统计
    memcpy(&p_st_report->st_l4_trans, p_st_l4_trans, sizeof(L4_transaction));
    //拷贝相关链表
    p_st_report->p_report_comm = FRREPORT_BODY_L7LIST_CLONE_COMM(p_st_private->p_st_body_data_req);
    if (p_st_report->p_report_comm == NULL)
        goto ERROR;
    p_st_report->p_report_req = FRREPORT_BODY_L7LIST_CLONE_REQ(p_st_private->p_st_body_data_req);
    if (p_st_report->p_report_req == NULL)
        goto ERROR;

    p_st_node = frlist_node_create(p_st_report);
    //存list之前 判断key值(mesgID)是否存在 如存在删除节点 再加节点
    p_st_gotnode = frhash2_getLong(p_st_private->p_rocketmq_list, i_opaque);
    if (p_st_gotnode != NULL)
    {
        frhash2_del(p_st_private->p_rocketmq_list, p_st_gotnode, rocketmq_list_node_free);
        p_st_gotnode = NULL;
    }
    frhash2_addLong(p_st_private->p_rocketmq_list, i_opaque, p_st_node);

    return FR_SUCC;

ERROR:
    if (p_st_report)
        frfree(p_st_report);
    if (p_st_report->p_report_comm)
        frreport_body_l7list_free(&p_st_report->p_report_comm);
    if (p_st_report->p_report_req)
        frreport_body_l7list_free(&p_st_report->p_report_req);
    return APM_L7_ANALYZE_FAIL;
}

/*!
 * 检查是否超时
 *
 * \param p_node
 * \param p_data
 *
 * \return int
 *
 * \note
 *
 */
FR_STATIC int __check_timeout(Frlist_node* p_node, const void *p_data)
{
    rocketmq_data *p_st_info = NULL;
    Frhash_header *p_header = (Frhash_header *)p_data;

    p_st_info = (rocketmq_data *)p_node->data;

    if (p_st_info->st_l4_trans.st_request_time.tv_sec + ROCKETMQ_TIMEOUT <= g_st_time.tv_sec)
    {
        frhash2_del(p_header, p_node, rocketmq_list_node_free);
        return 1;
    }

    return 1;
}

/*!
 * 清除hash中超时的node
 *
 * \param p_hash_stmt
 * \param st_timestamp
 *
 * \return int
 *
 * \note
 *
 */
FR_STATIC void __clear_timeout_node(Frhash_header *p_header, struct timeval st_timestamp)
{
    g_st_time = st_timestamp;
    frhash2_run(p_header, p_header, __check_timeout);

    return;
}

/*!
 * rocketmq的packet解析
 *
 * \param p_st_conn          (入力链接信息)
 * \param p_st_packet        (入力packet信息)
 * \param p_st_hash_report   (保存Frreport_data)
 * \param p_st_session       (私有数据)
 *
 * \return APM_L7_ANALYZE_SUCC   成功\n
 *         APM_L7_ANALYZE_FAIL   失败
 *         APM_L7_ANALYZE_TODO
 *
 * \note
 *
 */
FR_STATIC int __req_data_analysis(Conntrack *p_st_conn
                                , Apm_packet *p_st_packet
                                , Frhash_header *p_st_hash_report
                                , L7_session *p_st_session)
{
    int i_ret = 0;
    int i_opaque = 0;
    uint32_t u32_len = 0;
    unsigned long ul_tmp = 0;
    L4_transaction *p_st_l4_trans = NULL;
    Frreport_data *p_st_report_data = NULL;
    L7session_rocketmq_private *p_st_private = NULL;

    if (p_st_conn == NULL || p_st_packet == NULL || p_st_hash_report == NULL || p_st_session == NULL)
        return APM_L7_ANALYZE_FAIL;

    p_st_private = (L7session_rocketmq_private*)p_st_session->p_session_private;
    p_st_l4_trans = &(p_st_private->st_l4_trans);

    //创建数据链表
    if (p_st_private->p_st_body_data_req == NULL)
    {
        p_st_private->p_st_body_data_req = frreport_body_new_l7(p_st_conn);
        if (p_st_private->p_st_body_data_req == NULL)
            goto ERROR;

        if (frreport_body_l7_add_connInfo(p_st_private->p_st_body_data_req, p_st_conn) == FR_FAIL)
            goto ERROR;

        //seq ++
        p_st_private->ul_seq = frct_seq_increase(p_st_conn);

        if (p_st_l4_trans->st_request_time_first_byte.tv_sec==0)
        {
            //st_request_time_first_byte
            p_st_l4_trans->st_request_time_first_byte.tv_sec = p_st_packet->st_timestamp.tv_sec;
            p_st_l4_trans->st_request_time_first_byte.tv_usec = p_st_packet->st_timestamp.tv_usec;
        }

        //创建bytebuff
        if (p_st_private->p_req_packet == NULL)
        {
            p_st_private->p_req_packet = bytebuff_creat();
            if (p_st_private->p_req_packet == NULL)
                goto ERROR;
        }
    }
    //请求的字节数
    p_st_l4_trans->l_request_bytes += p_st_packet->ui_payload_len;

    if (p_st_private->p_req_packet == NULL)
        goto ERROR;

    //把这个包的数据先全加进bytebuff里
    if (bytebuff_add(p_st_private->p_req_packet, p_st_packet->p_payload, p_st_packet->ui_payload_len) != FR_SUCC)
        goto ERROR;

    if (bytebuff_get_int(p_st_private->p_req_packet, 4, &u32_len, BYTE_BIG_ENDIAN) == FR_FAIL)
        goto ERROR;
    bytebuff_rewind(p_st_private->p_req_packet);
    if (u32_len+4 > p_st_private->p_req_packet->ui_all_len)
        return APM_L7_ANALYZE_TODO;

    //req解析
    i_ret = __req_parse(p_st_private->p_req_packet, p_st_private->p_st_body_data_req, &i_opaque);
    if (i_ret == APM_L7_ANALYZE_FAIL)
        goto ERROR;

    if (frreport_body_l7_add_ulong(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_ID, p_st_private->ul_seq) == FR_FAIL)
        goto ERROR;

    //start time
    if (frreport_body_l7_add_timeval(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_START_TIMESTAMP, &p_st_l4_trans->st_request_time_first_byte) == FR_FAIL)
        goto ERROR;

    //st_request_time
    p_st_l4_trans->st_request_time.tv_sec = p_st_packet->st_timestamp.tv_sec;
    p_st_l4_trans->st_request_time.tv_usec = p_st_packet->st_timestamp.tv_usec;

    ul_tmp = frtime_diff(&(p_st_l4_trans->st_request_time_first_byte), &(p_st_l4_trans->st_request_time));
    //send_time_us
    if (frreport_body_l7_add_long(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_SEND_TIME_US, ul_tmp) == FR_FAIL)
        goto ERROR;

    //request bytes
    if (frreport_body_l7_add_long(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_DATA_BYTES, p_st_l4_trans->l_request_bytes) == FR_FAIL)
        goto ERROR;

    //end time
    if (frreport_body_l7_add_timeval(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_END_TIMESTAMP, &p_st_l4_trans->st_request_time) == FR_FAIL)
        goto ERROR;

    p_st_report_data = frreport_data_l7_new(FRREPORT_TYPE_L7DATA_REQ);
    if (p_st_report_data == NULL)
        goto ERROR;

    if (__rocketmq_add_req_list(p_st_private, i_opaque) == FR_FAIL)
        goto ERROR;

    __clear_timeout_node(p_st_private->p_rocketmq_list, p_st_packet->st_timestamp);

    if (frreport_data_add_body(p_st_report_data, p_st_private->p_st_body_data_req) == FR_FAIL)
        goto ERROR;
    p_st_private->p_st_body_data_req = NULL;

    if (frreport_data_hash(p_st_hash_report, p_st_report_data) == FR_FAIL)
        goto ERROR;
    p_st_report_data = NULL;

    __rocketmq_session_reset((void *)(p_st_private));

    return i_ret;

ERROR:
    if (p_st_report_data)
        frreport_data_free(&p_st_report_data);
    __rocketmq_session_reset((void *)(p_st_private));
    return APM_L7_ANALYZE_FAIL;
}

/*!
 * rocketmq的packet解析
 *
 * \param p_st_conn          (入力链接信息)
 * \param p_st_packet        (入力packet信息)
 * \param p_st_hash_report   (保存Frreport_data)
 * \param p_st_session       (私有数据)
 *
 * \return APM_L7_ANALYZE_SUCC   成功\n
 *         APM_L7_ANALYZE_FAIL   失败
 *         APM_L7_ANALYZE_TODO
 *
 * \note
 *
 */
FR_STATIC int __resp_data_analysis(Conntrack *p_st_conn
                                 , Apm_packet *p_st_packet
                                 , Frhash_header *p_st_hash_report
                                 , L7_session *p_st_session)
{
    int i_ret = 0;
    int i_opaque = 0;
    uint32_t u32_len = 0;
    unsigned long ul_tmp = 0;
    L4_transaction *p_st_l4_trans = NULL;
    Frreport_data* p_st_report_data = NULL;
    L7session_rocketmq_private *p_st_private = NULL;
    Frlist_node *p_st_node = NULL;
    rocketmq_data *p_st_data = NULL;

    if (p_st_conn == NULL || p_st_packet == NULL || p_st_hash_report == NULL || p_st_session == NULL)
        return APM_L7_ANALYZE_FAIL;

    p_st_private = (L7session_rocketmq_private*)p_st_session->p_session_private;
    p_st_l4_trans = &(p_st_private->st_l4_trans);

    if (p_st_private->p_st_body_data_resp == NULL)
    {
        //创建数据链表
        p_st_private->p_st_body_data_resp = frreport_body_new_l7(p_st_conn);
        if (p_st_private->p_st_body_data_resp == NULL)
            goto ERROR;

        if (p_st_l4_trans->st_response_time_first_byte.tv_sec == 0 && p_st_l4_trans->st_response_time_first_byte.tv_usec == 0)
        {
            p_st_l4_trans->st_response_time_first_byte.tv_sec = p_st_packet->st_timestamp.tv_sec;
            p_st_l4_trans->st_response_time_first_byte.tv_usec = p_st_packet->st_timestamp.tv_usec;
        }

        //创建bytebuff
        if (p_st_private->p_resp_packet == NULL)
        {
            p_st_private->p_resp_packet = bytebuff_creat();
            if (p_st_private->p_resp_packet == NULL)
                goto ERROR;
        }
    }

    //请求的字节数
    p_st_l4_trans->l_response_bytes += p_st_packet->ui_payload_len;

    if (p_st_private->p_resp_packet == NULL)
        goto ERROR;

    //把这个包的数据先全加进bytebuff里
    if (bytebuff_add(p_st_private->p_resp_packet, p_st_packet->p_payload, p_st_packet->ui_payload_len) != FR_SUCC)
        goto ERROR;

    if (bytebuff_get_int(p_st_private->p_resp_packet, 4, &u32_len, BYTE_BIG_ENDIAN) == FR_FAIL)
        goto ERROR;
    bytebuff_rewind(p_st_private->p_resp_packet);
    if (u32_len+4 > p_st_private->p_resp_packet->ui_all_len)
        return APM_L7_ANALYZE_TODO;

    //resp解析
    i_ret = __resp_parse(p_st_private->p_resp_packet, p_st_private->p_st_body_data_resp, &i_opaque);
    if (i_ret == APM_L7_ANALYZE_FAIL)
        goto ERROR;

    //取链表对应的request p_report_req p_report_comm
    p_st_node = frhash2_getLong(p_st_private->p_rocketmq_list, i_opaque);
    if (p_st_node == NULL)
        goto ERROR;
    p_st_data = (rocketmq_data *)p_st_node->data;
    //数据上报
    FRREPORT_BODY_L7LIST_SET_REQ(p_st_private->p_st_body_data_resp, &p_st_data->p_report_req);
    FRREPORT_BODY_L7LIST_SET_COMM(p_st_private->p_st_body_data_resp, &p_st_data->p_report_comm);

    //server time
    ul_tmp = frtime_diff(&p_st_data->st_l4_trans.st_request_time, &(p_st_l4_trans->st_response_time_first_byte));
    frhash2_del(p_st_private->p_rocketmq_list, p_st_node, rocketmq_list_node_free);
    if (frreport_body_l7_add_ulong(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_SERVER_TIME_US, ul_tmp) == FR_FAIL)
        goto ERROR;

    //start time
    if (frreport_body_l7_add_timeval(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_START_TIMESTAMP, &p_st_l4_trans->st_response_time_first_byte) == FR_FAIL)
        goto ERROR;

    //st_response_time
    p_st_l4_trans->st_response_time.tv_sec = p_st_packet->st_timestamp.tv_sec;
    p_st_l4_trans->st_response_time.tv_usec = p_st_packet->st_timestamp.tv_usec;

    if (frreport_body_l7_add_long(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_DATA_BYTES, p_st_l4_trans->l_response_bytes) == FR_FAIL)
        goto ERROR;

    //down load time
    ul_tmp = frtime_diff(&(p_st_l4_trans->st_response_time_first_byte), &(p_st_l4_trans->st_response_time));
    if (frreport_body_l7_add_long(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_DOWNLOAD_TIME_US, ul_tmp) == FR_FAIL)
        goto ERROR;

    //end time
    if (frreport_body_l7_add_timeval(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_END_TIMESTAMP, &p_st_l4_trans->st_response_time) == FR_FAIL)
        goto ERROR;

    p_st_report_data = frreport_data_l7_new(FRREPORT_TYPE_L7DATA_RESP);
    if (p_st_report_data == NULL)
        goto ERROR;

    if (frreport_data_add_body(p_st_report_data, p_st_private->p_st_body_data_resp) == FR_FAIL)
        goto ERROR;
    p_st_private->p_st_body_data_resp = NULL;

    if (frreport_data_hash(p_st_hash_report, p_st_report_data) == FR_FAIL)
        goto ERROR;
    p_st_report_data = NULL;

    __rocketmq_session_reset(p_st_session->p_session_private);

    return i_ret;

ERROR:
    if (p_st_report_data)
        frreport_data_free(&p_st_report_data);
    rocketmq_session_destroy(p_st_session->p_session_private);
    p_st_session->p_session_private = NULL;
    return APM_L7_ANALYZE_FAIL;
}

/*!
 * rocketmq的packet解析
 *
 * \param p_st_conn          (入力链接信息)
 * \param p_st_packet        (入力packet信息)
 * \param p_st_hash_report   (保存Frreport_data)
 *
 * \return APM_L7_ANALYZE_SUCC   成功\n
 *         APM_L7_ANALYZE_FAIL   失败
 *
 * \note
 *
 */
int rocketmq_analysis(Conntrack *p_st_conn
                           , Apm_packet *p_st_packet
                           , Frhash_header *p_st_hash_report)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    L7_session *p_st_session = NULL;

    if (p_st_conn == NULL || p_st_packet == NULL || p_st_hash_report == NULL)
        return APM_L7_ANALYZE_FAIL;

    //到Conntrack中取得session信息
    p_st_session = frct_get_session(p_st_conn);
    if (p_st_session == NULL)
        return APM_L7_ANALYZE_FAIL;

    //fin、syn包销毁私有数据
    if (p_st_packet->em_analyz_cmd == FRANALYZ_CMD_FIN || p_st_packet->em_analyz_cmd == FRANALYZ_CMD_SYN)
    {
        rocketmq_session_destroy(p_st_session->p_session_private);
        p_st_session->p_session_private = NULL;
        return APM_L7_ANALYZE_SUCC;
    }

    //p_session_private分配内存
    if (p_st_session->p_session_private == NULL)
    {
        p_st_session->p_session_private = __rocketmq_private_create();
        //内存分配失败
        if (p_st_session->p_session_private == NULL)
            return APM_L7_ANALYZE_FAIL;
    }

    //request
    if (p_st_packet->i_direction == FR_DIRECT_CLIENT2SRV)
    {
        i_ret = __req_data_analysis(p_st_conn, p_st_packet, p_st_hash_report, p_st_session);
        if (i_ret == APM_L7_ANALYZE_FAIL)
            goto ERROR;
        return i_ret;
    }
    //response
    else
    {
        i_ret = __resp_data_analysis(p_st_conn, p_st_packet, p_st_hash_report, p_st_session);
        if (i_ret == APM_L7_ANALYZE_FAIL)
            goto ERROR;
        return i_ret;
    }

ERROR:
    rocketmq_session_destroy(p_st_session->p_session_private);
    p_st_session->p_session_private = NULL;
    return APM_L7_ANALYZE_FAIL;
}

/*!
 * 私有信息销毁
 *
 * \param p_private_session  (rocketmq的私有session数据)
 *
 * \return FR_SUCC   成功\n
 *
 * \note
 *
 */
int rocketmq_session_destroy(void *p_private_session)
{
    L7session_rocketmq_private *p_st_private = (L7session_rocketmq_private*)p_private_session;

    if (p_st_private)
    {
        if (p_st_private->p_st_body_data_req)
            frreport_body_free(&p_st_private->p_st_body_data_req);

        if (p_st_private->p_st_body_data_resp)
            frreport_body_free(&p_st_private->p_st_body_data_resp);

        if (p_st_private->p_req_packet)
            bytebuff_free(&(p_st_private->p_req_packet));

        if (p_st_private->p_resp_packet)
            bytebuff_free(&(p_st_private->p_resp_packet));

        memset(&p_st_private->st_l4_trans, 0x00, sizeof(L4_transaction));

        frhash2_free2(p_st_private->p_rocketmq_list, rocketmq_list_node_free);

        frfree(p_st_private);
    }

    return FR_SUCC;
}
